/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 上午11:17:52
*/
package cn.com.guage.flink.transformation;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * @version:1.0.1
 * @Description: （对类进行功能描述）
 * @author: yangdechao
 * @date: datedate 2021年11月15日 上午11:17:52
 */
public class CoFlatMapFunctionTest {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		DataStreamSource<String>  dataStream1 = env.fromElements("I love you","my name is guage");
		DataStreamSource<String> dataStream2 = env.fromElements("Tom", "Jack", "Marry");
		
		/**
		 * 对于ConnectedStreams，我们需要重写CoMapFunction或CoFlatMapFunction。
		 * 这两个接口都提供了三个泛型，这三个泛型分别对应第一个输入流的数据类型、
		 * 第二个输入流的数据类型和输出流的数据类型。在重写函数时，对于CoMapFunction，
		 * map1处理第一个流的数据，map2处理第二个流的数据；对于CoFlatMapFunction，
		 * flatMap1处理第一个流的数据，flatMap2处理第二个流的数据。Flink并不能保证两个函数调用顺序，
		 * 两个函数的调用依赖于两个数据流数据的流入先后顺序，即第一个数据流有数据到达时，
		 * map1或flatMap1会被调用，第二个数据流有数据到达时，map2或flatMap2会被调用。
		 * 下面的代码对一个整数流和一个字符串流进行了connect操作。
		 */
		dataStream1.connect(dataStream2).flatMap(new CoFlatMapFunction<String, String, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 3076485499068113930L;

			@Override
			public void flatMap1(String value, Collector<String> out) throws Exception {
				String [] splits = value.split(" ");
				for(String split : splits) {
					out.collect(split);
				}
			}

			@Override
			public void flatMap2(String value, Collector<String> out) throws Exception {
				out.collect(value);
			}
		}).print();
		env.execute();
	}
}
